package pers.wang.toolset.ali.rocketmq.spring.boot.autoconfigure;

import java.util.Properties;

import javax.annotation.Resource;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Conditional;
import org.springframework.context.annotation.Configuration;

import com.aliyun.openservices.ons.api.Consumer;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.Producer;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.transaction.TransactionProducer;

import pers.wang.toolset.ali.rocketmq.spring.boot.autoconfigure.condition.EnableConsumerCondition;
import pers.wang.toolset.ali.rocketmq.spring.boot.autoconfigure.condition.EnableProducerCondition;
import pers.wang.toolset.ali.rocketmq.spring.boot.autoconfigure.condition.EnableRocketMqCondition;

@Configuration
@EnableConfigurationProperties({ProducerConfig.class,ConsumerConfig.class})
@Conditional({
	EnableRocketMqCondition.class
})
public class RocketMqConfigure {
	private Logger LOG = LoggerFactory.getLogger(RocketMqConfigure.class);
	@Resource
	private ProducerConfig producerConfig;
	@Resource
	private ConsumerConfig consumerConfig;
	
	@Bean
	@Conditional(EnableProducerCondition.class)
	public TransactionProducer transactionProducer() {
		if(producerConfig.getTransaction()) {
			LOG.info("创建阿里云Rocket 事务消息Producer");
			TransactionProducer producer = ONSFactory.createTransactionProducer(getProducerProperties(), transactionChecker());
			producer.start();
			return producer;
		}
		return null;
	}
	@Bean
	@Conditional(EnableProducerCondition.class)
	public Producer producer() {
		if(!producerConfig.getTransaction()) {
			LOG.info("创建阿里云Rocket 普通消息Producer");
			Producer producer = ONSFactory.createProducer(getProducerProperties());
			producer.start();
			return producer;
		}
		return null;
	}
	@Bean
	@ConditionalOnProperty(name = "toolset.ali.rocketmq.producer.transaction",havingValue = "true")
	public TransactionChecker transactionChecker() {
		return new TransactionChecker();
	}
	@Bean
	@Conditional(EnableConsumerCondition.class)
	public Consumer consumer() {
		LOG.info("创建消息消费者");
		Consumer consumer = ONSFactory.createConsumer(getConsumerProperties());
		if(consumerConfig.getTopics() == null  || consumerConfig.getTopics().length == 0) {
			throw new Error("消费者topic为空");
		}
		
		for(String topic : consumerConfig.getTopics()) {
			consumer.subscribe(topic, "", messageListener());
		}
		consumer.start();
		return consumer;
	}
	
	@Bean
	@Conditional(EnableConsumerCondition.class)
	public MessageListener messageListener() {
		return new MessageListener();
	}
	
	private  Properties getProducerProperties() {
		Properties pro = new Properties();
//		pro.put(PropertyKeyConst.GROUP_ID, producerConfig.getGroupId());
		pro.put(PropertyKeyConst.AccessKey, producerConfig.getAccessKey());
		pro.put(PropertyKeyConst.SecretKey, producerConfig.getSecretKey());
		pro.put(PropertyKeyConst.NAMESRV_ADDR, producerConfig.getNamesrvAddr());
		return pro;
	}
	
	private  Properties getConsumerProperties() {
		Properties pro = new Properties();
		pro.put(PropertyKeyConst.GROUP_ID, consumerConfig.getGroupId());
		pro.put(PropertyKeyConst.AccessKey, consumerConfig.getAccessKey());
		pro.put(PropertyKeyConst.SecretKey, consumerConfig.getSecretKey());
		pro.put(PropertyKeyConst.NAMESRV_ADDR, consumerConfig.getNamesrvAddr());
		return pro;
	}
}
